package com.baishancloud.log

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.util.DimMessageFetch
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.slf4j.{Logger, LoggerFactory}

import java.util.{Timer, TimerTask}
import scala.collection.mutable

/**
 * 将结果对象中的域名字符串替换为域名id，最后输出字符串。
 *
 * @author ziqiang.wang
 * @date 2021/12/6 16:51
 */
class ResultParse() extends RichMapFunction[MiguICTAuditLog, String] {

  val LOGGER: Logger = LoggerFactory.getLogger(classOf[ResultParse])
  val domainIds: mutable.Map[String, Int] = mutable.Map[String, Int]()
  var subTaskIndex: Int = -1

  //半小时刷新一次域名信息
  override def open(parameters: Configuration): Unit = {
    subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    DimMessageFetch.fetchDomainIdAll(domainIds)
    LOGGER.info(s"domainIds${subTaskIndex}初始化完成")
    val timer = new Timer("update DomainIds", true)
    timer.scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = DimMessageFetch.fetchDomainIdAll(domainIds)
    }, 1800 * 1000, 1800 * 1000)
  }

  override def map(value: MiguICTAuditLog): String = {
    JSONUtil.toJsonStr(MiguICTAuditResult(value.timeAt, domainIds.getOrElse(value.domain, -9999), value.line, value.traffic))
  }
}
